//package io.github.talelin.latticy.mq.consumer;
//
//import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
//import com.rabbitmq.client.Channel;
//import io.github.talelin.latticy.bo.StockBO;
//import io.github.talelin.latticy.bo.StockOnTheWayBO;
//import io.github.talelin.latticy.common.constant.Constant;
//import io.github.talelin.latticy.model.MsgLogDO;
//import io.github.talelin.latticy.model.StockDO;
//import io.github.talelin.latticy.mq.MessageHelper;
//import io.github.talelin.latticy.service.StockService;
//import io.github.talelin.latticy.service.feign.OmsMsgLogFeignClient;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.springframework.amqp.core.Message;
//import org.springframework.amqp.core.MessageProperties;
//import org.springframework.amqp.rabbit.annotation.RabbitListener;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.http.HttpRequest;
//import org.springframework.stereotype.Component;
//
//import java.io.IOException;
//import java.util.List;
//
//@Component
//@Slf4j
//public class SimpleStockConsumer {
//
//    @Autowired
//    private StockService stockService;
//    @Autowired
//    private OmsMsgLogFeignClient omsMsgLogFeignClient;
//
//    @RabbitListener(queues = "inbound.order.no.queue")
//    public void consume(Message message, Channel channel, HttpRequest httpRequest ) throws IOException {
//        try {
//            StockOnTheWayBO stockOnTheWayBO = MessageHelper.msgToObj(message, StockOnTheWayBO.class);
//            log.info("收到消息: {}", stockOnTheWayBO.toString());
//            Long msgId = stockOnTheWayBO.getMsgId();
//            List<StockBO> list = stockOnTheWayBO.getList();
//            MsgLogDO msgLog = omsMsgLogFeignClient.getMsgLogById(msgId);
//            if (null == msgLog || msgLog.getStatus().equals(Constant.MsgLogStatus.CONSUMED_SUCCESS)) {// 消费幂等性
//                log.info("重复消费, msgId: {}", msgId);
//                return;
//            }
//
//            MessageProperties properties = message.getMessageProperties();
//            long tag = properties.getDeliveryTag();
//
//            list.parallelStream().forEach(
//                    stockBO -> {
//                        UpdateWrapper<StockDO> wrapper = new UpdateWrapper();
//                        wrapper.lambda().eq(StockDO::getWarehouseCode, stockBO.getWarehouseCode()).eq(StringUtils.isNotBlank(stockBO.getTenant()),StockDO::getTenant, stockBO.getTenant())
//                                .eq(StockDO::getSku, stockBO.getSku()).setSql("on_the_way = on_the_way +" + stockBO.getOnTheWay());
//                        boolean success = stockService.saveOrUpdate(new StockDO(stockBO), wrapper);
//                        try {
//                            if (success) {
//                                msgLog.setStatus(Constant.MsgLogStatus.CONSUMED_SUCCESS);
//                                // 验证定时任务的消息重投 下面这行代码注释
//                                omsMsgLogFeignClient.updateById(msgLog);
//                                channel.basicAck(tag, false);// 消费确认
//                            } else {
//                                channel.basicNack(tag, false, true);
//                            }
//                        }catch (Exception e){
//                            e.printStackTrace();
//                        }
//
//                    }
//
//            );
//        }catch (Exception e){
//            e.printStackTrace();
//        }
//    }
//
//}
